package com.shujia.spark.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo22Broadcast {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("spark")


    val sc = new SparkContext(conf)


    val students: RDD[String] = sc.textFile("data/students.txt")

    /*
        val ids = List("1500100010", "1500100013", "1500100015", "1500100016")


        val filterRDD: RDD[String] = students.filter(student => {

          val id: String = student.split(",")(0)

          ids.contains(id)
        })

        filterRDD.foreach(println)*/


    /**
      * 广播变量
      *
      */


    val ids = List("1500100010", "1500100013", "1500100015", "1500100016")

    //1、在Driver端 将一个变量广播出去
    val broIds: Broadcast[List[String]] = sc.broadcast(ids)


    val filterRDD: RDD[String] = students.filter(student => {

      val id: String = student.split(",")(0)

      //2、在Executor使用广播变量
      val value: List[String] = broIds.value

      value.contains(id)
    })

    filterRDD.foreach(println)


    /**
      * 广播变量的应用
      *
      *
      * 实现map join
      * 将小表加载内存中，在map端进行关联
      *
      */

    val students1: RDD[String] = sc.textFile("data/students.txt")
    val scores: RDD[String] = sc.textFile("data/score.txt")


    /**
      * collect : 将rdd的数据拉去到Driver端的内存中
      *
      */
    val list: Array[String] = students1.collect()

    val studentMap: Map[String, String] = list.map(stu => {
      val id: String = stu.split(",")(0)
      (id, stu)
    }).toMap

    //将小表广播
    val broStudentMap: Broadcast[Map[String, String]] = sc.broadcast(studentMap)


    val stuCoInfo: RDD[String] = scores.map(sco => {
      val id: String = sco.split(",")(0)

      //读取广播变量
      val value: Map[String, String] = broStudentMap.value

      //使用id 到学生表的map中获取学生信息
      val studentInfo: String = value.getOrElse(id, "默认值")

      studentInfo + "\t" + sco
    })

    stuCoInfo.foreach(println)


  }

}
